package com.sui.bigdata.flink.sql.function.aggregate;

import com.sui.bigdata.flink.sql.function.util.DuplicateRemovalUtils;
import org.apache.flink.table.functions.AggregateFunction;
import com.sui.bigdata.flink.sql.function.scalar.CurrentTime;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Iterator;

/**
 * @author LiuHuan
 * @date 2019/9/19 21:56
 * @description
 * @reviewer
 */
public class DuplicateRemoval extends AggregateFunction<Integer, DuplicateRemovalUtils> {
    @Override
    public DuplicateRemovalUtils createAccumulator() {
        return new DuplicateRemovalUtils();
    }

    @Override
    public Integer getValue(DuplicateRemovalUtils duplicateRemovalUtils) {
        Integer result = 0;
        if (!(duplicateRemovalUtils == null || duplicateRemovalUtils.getCount() == 0)){
            result = duplicateRemovalUtils.getCount();
        }
        return result;
    }

    public void accumulate(DuplicateRemovalUtils duplicateRemovalUtils, String productCode, String productType, String field, String ymd, Integer interval) throws ParseException {
        CurrentTime currentTime = new CurrentTime();
        try {
            if (productCode != null && ymd != null && !duplicateRemovalUtils.exitsProductCode(productCode) && field.equals(productType) && daysBetween(ymd,currentTime.eval(interval)) <= 0){
                duplicateRemovalUtils.addProductCode(productCode);
                duplicateRemovalUtils.setCount(duplicateRemovalUtils.getCount() + 1);
            }
        }catch (Exception e){}
    }

    public void accumulate(DuplicateRemovalUtils duplicateRemovalUtils, String productCode, String productType, String field, String approveStatus, String field2, String ymd, Integer interval) throws ParseException {
        CurrentTime currentTime = new CurrentTime();
        try {
            if (productCode != null && ymd != null && !duplicateRemovalUtils.exitsProductCode(productCode) && field.equals(productType) && field2.equals(approveStatus) && daysBetween(ymd,currentTime.eval(interval)) <= 0){
                duplicateRemovalUtils.addProductCode(productCode);
                duplicateRemovalUtils.setCount(duplicateRemovalUtils.getCount() + 1);
            }
        }catch (Exception e){}
    }

    public int daysBetween(String ymd, String interval) throws ParseException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        Calendar cal = Calendar.getInstance();
        cal.setTime(sdf.parse(ymd));
        long ymdFormat = cal.getTimeInMillis();
        cal.setTime(sdf.parse(interval));
        long intervalFormat = cal.getTimeInMillis();
        long betweenDays = (intervalFormat - ymdFormat) / 86400000L;
        return Integer.parseInt(String.valueOf(betweenDays));
    }

    public void merge(DuplicateRemovalUtils duplicateRemovalUtils, Iterable<DuplicateRemovalUtils> it) {
        Iterator<DuplicateRemovalUtils> iter = it.iterator();
        while (iter.hasNext()) {
            DuplicateRemovalUtils a = iter.next();
            duplicateRemovalUtils.setGroups(a.getGroups());
            duplicateRemovalUtils.setCount(a.getCount());
            duplicateRemovalUtils.setRepetition(a.getRepetition());
        }
    }
}
